package org.wyy.sample;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;

/**
 * @author mars
 * @date 2024/1/22
 **/
public class Barrier extends SyncPrimitive{

    int size;
    String clientHostname;

    /**
     * 构造器
     * 连接zk后,该类实例将成为zk的客户端
     *
     * @param address zk地址
     * @param root 根节点
     * @param size 屏障大小
     */
    Barrier(String address, String root, int size) {
        super(address);
        this.root = root;
        this.size = size;

        // 构建屏障节点
        if (zk != null) {
            try {
                Stat s = zk.exists(root, false);
                if (s == null) {
                    zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
                            CreateMode.PERSISTENT);
                }
            } catch (KeeperException e) {
                System.out.println("Keeper exception when instantiating Barrier: " + e);
            } catch (InterruptedException e) {
                System.out.println("Interrupted exception");
            }
        }

        // 设置节点名称
        try {
            clientHostname = InetAddress.getLocalHost().getCanonicalHostName();
        } catch (UnknownHostException e) {
            System.out.println(e);
        }

    }

    /**
     * 加入屏障
     *
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    boolean enter() throws KeeperException, InterruptedException{
        // 创建临时节点
        String nodeName = zk.create(root + "/" + clientHostname, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("创建zkClient节点: " + nodeName);
        while (true) {
            synchronized (mutex) {
                // 获取root节点下所有子节点,并监听
                List<String> list = zk.getChildren(root, true);
                if (list.size() < size) {
                    // 达到屏障阈值前,等待process回调函数唤醒
                    mutex.wait();
                } else {
                    System.out.println("达到屏障容量阈值");
                    return true;
                }
            }
        }
    }

    /**
     * Wait until all reach barrier
     *
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    boolean leave() throws KeeperException, InterruptedException{
        zk.delete(root + "/" + clientHostname, 0);
        while (true) {
            synchronized (mutex) {
                List<String> list = zk.getChildren(root, true);
                if (list.size() > 0) {
                    //等待process回调函数唤醒
                    mutex.wait();
                } else {
                    return true;
                }
            }
        }
    }
}
